/*

 * Licensed to the Apache Software Foundation (ASF) under one

 * or more contributor license agreements.  See the NOTICE file

 * distributed with this work for additional information

 * regarding copyright ownership.  The ASF licenses this file

 * to you under the Apache License, Version 2.0 (the

 * "License"); you may not use this file except in compliance

 * with the License.  You may obtain a copy of the License at

 *

 *     http://www.apache.org/licenses/LICENSE-2.0

 *

 * Unless required by applicable law or agreed to in writing, software

 * distributed under the License is distributed on an "AS IS" BASIS,

 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

 * See the License for the specific language governing permissions and

 * limitations under the License.

 */

package com.bff.gaia.unified.runners.core;



import com.google.auto.value.AutoValue;

import com.bff.gaia.unified.sdk.coders.Coder;

import com.bff.gaia.unified.sdk.coders.CoderException;

import com.bff.gaia.unified.sdk.coders.InstantCoder;

import com.bff.gaia.unified.sdk.coders.StringUtf8Coder;

import com.bff.gaia.unified.sdk.coders.StructuredCoder;

import com.bff.gaia.unified.sdk.state.TimeDomain;

import com.bff.gaia.unified.sdk.transforms.DoFn;

import com.bff.gaia.unified.sdk.transforms.windowing.BoundedWindow;

import com.bff.gaia.unified.vendor.guava.com.google.common.collect.ComparisonChain;

import org.joda.time.Instant;



import javax.annotation.Nullable;

import java.io.IOException;

import java.io.InputStream;

import java.io.OutputStream;

import java.util.Arrays;

import java.util.List;



/**

 * Encapsulate interaction with time within the execution environment.

 *

 * <p>This class allows setting and deleting timers, and also retrieving an estimate of the current

 * time.

 */

public interface TimerInternals {



  /**

   * Sets a timer to be fired when the current time in the specified time domain reaches the target

   * timestamp.

   *

   * <p>The combination of {@code namespace} and {@code timerId} uniquely identify a timer.

   *

   * <p>If a timer is set and then set again before it fires, later settings should clear the prior

   * setting.

   *

   * <p>It is an error to set a timer for two different time domains.

   */

  void setTimer(StateNamespace namespace, String timerId, Instant target, TimeDomain timeDomain);



  /** @deprecated use {@link #setTimer(StateNamespace, String, Instant, TimeDomain)}. */

  @Deprecated

  void setTimer(TimerData timerData);



  /**

   * Deletes the given timer.

   *

   * <p>A timer's ID is enforced to be unique in validation of a {@link DoFn}, but runners often

   * manage timers for different time domains in very different ways, thus the {@link TimeDomain} is

   * a required parameter.

   */

  void deleteTimer(StateNamespace namespace, String timerId, TimeDomain timeDomain);



  /** @deprecated use {@link #deleteTimer(StateNamespace, String, TimeDomain)}. */

  @Deprecated

  void deleteTimer(StateNamespace namespace, String timerId);



  /** @deprecated use {@link #deleteTimer(StateNamespace, String, TimeDomain)}. */

  @Deprecated

  void deleteTimer(TimerData timerKey);



  /** Returns the current timestamp in the {@link TimeDomain#PROCESSING_TIME} time domain. */

  Instant currentProcessingTime();



  /**

   * Returns the current timestamp in the {@link TimeDomain#SYNCHRONIZED_PROCESSING_TIME} time

   * domain or {@code null} if unknown.

   */

  @Nullable

  Instant currentSynchronizedProcessingTime();



  /**

   * Return the current, local input watermark timestamp for this computation in the {@link

   * TimeDomain#EVENT_TIME} time domain.

   *

   * <p>This value:

   *

   * <ol>

   *   <li>Is never {@literal null}, but may be {@link BoundedWindow#TIMESTAMP_MIN_VALUE}.

   *   <li>Is monotonically increasing.

   *   <li>May differ between workers due to network and other delays.

   *   <li>Will never be ahead of the global input watermark for this computation. But it may be

   *       arbitrarily behind the global input watermark.

   *   <li>Any element with a timestamp before the local input watermark can be considered 'locally

   *       late' and be subject to special processing or be dropped entirely.

   * </ol>

   *

   * <p>Note that because the local input watermark can be behind the global input watermark, it is

   * possible for an element to be considered locally on-time even though it is globally late.

   */

  Instant currentInputWatermarkTime();



  /**

   * Return the current, local output watermark timestamp for this computation in the {@link

   * TimeDomain#EVENT_TIME} time domain. Return {@code null} if unknown.

   *

   * <p>This value:

   *

   * <ol>

   *   <li>Is monotonically increasing.

   *   <li>Will never be ahead of {@link #currentInputWatermarkTime} as returned above.

   *   <li>May differ between workers due to network and other delays.

   *   <li>However will never be behind the global input watermark for any following computation.

   * </ol>

   *

   * <p>In pictures:

   *

   * <pre>{@code

   *  |              |       |       |       |

   *  |              |   D   |   C   |   B   |   A

   *  |              |       |       |       |

   * GIWM     <=    GOWM <= LOWM <= LIWM <= GIWM

   * (next stage)

   * -------------------------------------------------> event time

   * }</pre>

   *

   * <p>where

   *

   * <ul>

   *   <li>LOWM = local output water mark.

   *   <li>GOWM = global output water mark.

   *   <li>GIWM = global input water mark.

   *   <li>LIWM = local input water mark.

   *   <li>A = A globally on-time element.

   *   <li>B = A globally late, but locally on-time element.

   *   <li>C = A locally late element which may still contribute to the timestamp of a pane.

   *   <li>D = A locally late element which cannot contribute to the timestamp of a pane.

   * </ul>

   *

   * <p>Note that if a computation emits an element which is not before the current output watermark

   * then that element will always appear locally on-time in all following computations. However, it

   * is possible for an element emitted before the current output watermark to appear locally

   * on-time in a following computation. Thus we must be careful to never assume locally late data

   * viewed on the output of a computation remains locally late on the input of a following

   * computation.

   */

  @Nullable

  Instant currentOutputWatermarkTime();



  /** Data about a timer as represented within {@link TimerInternals}. */

  @AutoValue

  abstract class TimerData implements Comparable<TimerData> {



    public abstract String getTimerId();



    public abstract StateNamespace getNamespace();



    public abstract Instant getTimestamp();



    public abstract TimeDomain getDomain();



    // When adding a new field, make sure to add it to the compareTo() method.



    /**

     * Construct a {@link TimerData} for the given parameters, where the timer ID is automatically

     * generated.

     */

    public static TimerData of(

		String timerId, StateNamespace namespace, Instant timestamp, TimeDomain domain) {

      return new AutoValue_TimerInternals_TimerData(timerId, namespace, timestamp, domain);

    }



    /**

     * Construct a {@link TimerData} for the given parameters, where the timer ID is

     * deterministically generated from the {@code timestamp} and {@code domain}.

     */

    public static TimerData of(StateNamespace namespace, Instant timestamp, TimeDomain domain) {

      String timerId =

          new StringBuilder()

              .append(domain.ordinal())

              .append(':')

              .append(timestamp.getMillis())

              .toString();

      return of(timerId, namespace, timestamp, domain);

    }



    /**

     * {@inheritDoc}.

     *

     * <p>Used for sorting {@link TimerData} by timestamp. Furthermore, we compare timers by all the

     * other fields so that {@code compareTo()} only returns 0 when {@code equals()} returns 0. This

     * ensures consistent sort order.

     */

    @Override

    public int compareTo(TimerData that) {

      if (this.equals(that)) {

        return 0;

      }

      ComparisonChain chain =

          ComparisonChain.start()

              .compare(this.getTimestamp(), that.getTimestamp())

              .compare(this.getDomain(), that.getDomain())

              .compare(this.getTimerId(), that.getTimerId());

      if (chain.result() == 0 && !this.getNamespace().equals(that.getNamespace())) {

        // Obtaining the stringKey may be expensive; only do so if required

        chain = chain.compare(getNamespace().stringKey(), that.getNamespace().stringKey());

      }

      return chain.result();

    }

  }



  /** A {@link Coder} for {@link TimerData}. */

  class TimerDataCoder extends StructuredCoder<TimerData> {

    private static final StringUtf8Coder STRING_CODER = StringUtf8Coder.of();

    private static final InstantCoder INSTANT_CODER = InstantCoder.of();

    private final Coder<? extends BoundedWindow> windowCoder;



    public static TimerDataCoder of(Coder<? extends BoundedWindow> windowCoder) {

      return new TimerDataCoder(windowCoder);

    }



    private TimerDataCoder(Coder<? extends BoundedWindow> windowCoder) {

      this.windowCoder = windowCoder;

    }



    @Override

    public void encode(TimerData timer, OutputStream outStream) throws CoderException, IOException {

      STRING_CODER.encode(timer.getTimerId(), outStream);

      STRING_CODER.encode(timer.getNamespace().stringKey(), outStream);

      INSTANT_CODER.encode(timer.getTimestamp(), outStream);

      STRING_CODER.encode(timer.getDomain().name(), outStream);

    }



    @Override

    public TimerData decode(InputStream inStream) throws CoderException, IOException {

      String timerId = STRING_CODER.decode(inStream);

      StateNamespace namespace =

          StateNamespaces.fromString(STRING_CODER.decode(inStream), windowCoder);

      Instant timestamp = INSTANT_CODER.decode(inStream);

      TimeDomain domain = TimeDomain.valueOf(STRING_CODER.decode(inStream));

      return TimerData.of(timerId, namespace, timestamp, domain);

    }



    @Override

    public List<? extends Coder<?>> getCoderArguments() {

      return Arrays.asList(windowCoder);

    }



    @Override

    public void verifyDeterministic() throws NonDeterministicException {

      verifyDeterministic(this, "window coder must be deterministic", windowCoder);

    }

  }

}